package com.dfd.consumer.listener;

import cn.hutool.core.collection.CollectionUtil;
import com.dfd.common.mq.po.MessageBody;
import com.dfd.consumer.service.MqRecordService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.nio.charset.StandardCharsets;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * 消费者监听器
 * Created on 2021/12/21.
 *
 * @author DD
 */
@Slf4j
@Component
public class AConsumerListener implements MessageListenerConcurrently {
    @Autowired
    MqRecordService mqRecordService;
    @Autowired
    RedisTemplate redisTemplate;
    @Autowired
    ThreadPoolExecutor threadPoolExecutor;

    /**
     * 默认msg里只有一条消息，可以通过设置consumeMessageBatchMaxSize参数来批量接收消息
     * 不要抛异常，如果没有return CONSUME_SUCCESS ，consumer会重新消费该消息，直到return CONSUME_SUCCESS
     *
     * @param msgList
     * @param consumeConcurrentlyContext
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgList, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
        if (CollectionUtils.isEmpty(msgList)) {
            log.info("MQ接收消息为空，直接返回成功");
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        List<List<MessageExt>> split = CollectionUtil.split(msgList, 100);
        CountDownLatch countDownLatch = new CountDownLatch(split.size());
        for (List<MessageExt> messageExts : split) {
            threadPoolExecutor.execute(() -> {
                for (MessageExt messageExt : messageExts) {

                    String topic = messageExt.getTopic();
                    String tags = messageExt.getTags();
                    String msgKey = messageExt.getKeys();
                    String content = new String(messageExt.getBody(), StandardCharsets.UTF_8);
                    MessageBody mb = MessageBody.builder().msgKey(msgKey).content(content).build();

                    // 直接放到Redis
//                    BoundSetOperations ops = redisTemplate.boundSetOps("source");
//                    ops.add(mb);
                    BoundHashOperations ops = redisTemplate.boundHashOps("rocketmq-log");
                    ops.putIfAbsent(msgKey, mb);
                    log.info("收到的消息：{},业务ID：{}，Topic：{}，Tag：{}，成功存入Redis", content, msgKey, topic, tags);
                }
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
